package com.corpgovernment.common.mq.consumer;

import com.corpgovernment.common.base.BaseService;
import com.ctrip.corp.obt.generic.utils.CollectionUtils;
import com.ctrip.corp.obt.generic.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.PostConstruct;

/**
 * @program: parent
 * @description: mq消费者抽象父类
 * @author: wubing
 * @create: 2019-07-03 15:09
 */
@Slf4j
public abstract class AbstractMQConsumer extends BaseService {

	@Value("${rocketmq.consumer.namesrvAddr:f}")
	private String namesrvAddr;
	@Value("${rocketmq.consumer.consumeThreadMin:20}")
	private int consumeThreadMin;
	@Value("${rocketmq.consumer.consumeThreadMax:64}")
	private int consumeThreadMax;
	@Value("${rocketmq.consumer.consumeMessageBatchMaxSize:1}")
	private int consumeMessageBatchMaxSize;
	@Value("${rocketmq.consumer.accesskey:}")
	private String accesskey;
	@Value("${rocketmq.consumer.accesskey-secret:}")
	private String accesskeySecret;

	@PostConstruct
	private void initConsumer() {
		if ("f".equals(this.namesrvAddr)) {
			log.warn("this server never used RocketMQ......");
			return;
		}
		AclClientRPCHook rpcHook = null;
		log.info("accesskey:{},accesskeySecret:{}", accesskey, accesskeySecret);
		if (StringUtils.isNotBlank(accesskey) && StringUtils.isNotBlank(accesskeySecret)) {
			rpcHook = new AclClientRPCHook(new SessionCredentials(accesskey, accesskeySecret));
		}
		final MQTopicConfig config = buildConfig();
		DefaultMQPushConsumer consumer;
		if (rpcHook != null){
			consumer = new DefaultMQPushConsumer(config.getGroupName(), rpcHook, new AllocateMessageQueueAveragely(), true, null);
			consumer.setAccessChannel(AccessChannel.CLOUD);
		}else{
			consumer = new DefaultMQPushConsumer(config.getGroupName());
		}
		// 设置为您从阿里云消息队列RocketMQ版控制台获取的接入点信息，类似“http://MQ_INST_XXXX.aliyuncs.com:80”。
		consumer.setNamesrvAddr(namesrvAddr);
		consumer.setConsumeThreadMin(consumeThreadMin);
		consumer.setConsumeThreadMax(consumeThreadMax);
		/**
		 * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
		 * 如果非第一次启动，那么按照上次消费的位置继续消费
		 */
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		/**
		 * 设置消费模型，集群还是广播，默认为集群
		 */
		//consumer.setMessageModel(MessageModel.CLUSTERING);
		/**
		 * 设置一次消费消息的条数，默认为1条
		 */
		consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
		consumer.registerMessageListener(consumer());
		try {
			consumer.subscribe(config.getTopic(), config.getTags());
			consumer.start();
		} catch (Exception e) {
			log.error("consumer is start fail! topics:{},namesrvAddr:{},打印异常:{}", config.getTopic(), namesrvAddr, e);

		}
		log.info("consumer is start success !topics:{},namesrvAddr:{}",
				 config.getTopic(), namesrvAddr);
	}

	protected MessageListenerConcurrently consumer() {
		final MQTopicConfig config = buildConfig();
		return (var1, var2) -> {
			if (!CollectionUtils.isEmpty(var1)) {
				for (MessageExt messageExt : var1) {
					if (messageExt.getTopic().equals(config.getTopic())) {
						try {
							consume(messageExt, var2);
						} catch (Exception e) {
							log.warn("消费topic:{},失败，消息体：{}，打印异常：{}", config.getTopic(), messageExt, e);
							return ConsumeConcurrentlyStatus.RECONSUME_LATER;
						}
					} else {
						// 如果没有return success, consumer会重新消费该消息, 直到return success
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}
				}
			}
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		};
	}

	/**
	 * 消费
	 *
	 * @param msg
	 * @param context
	 * @throws Exception
	 */
	protected abstract void consume(MessageExt msg, ConsumeConcurrentlyContext context) throws Exception;

    /**
     * 要消费的topic
     *
     * @return
     */
    protected abstract MQTopicConfig buildConfig();

}